import com.atguigu.gmall.realtime.app.BaseSQLAppV2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2023/6/26 16:23
 */
public class SQLDemo extends BaseSQLAppV2 {
    public static void main(String[] args) {
        new SQLDemo().init(2001, 2, "SQLDemo");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        SingleOutputStreamOperator<WaterSensor> stream = env.socketTextStream("hadoop162", 8888)
            .map(new WaterSensorMapFunction());
        
        Table table = tEnv.fromDataStream(stream);
        
        Table t1 = table
            .groupBy($("id"))
            .select($("id"), $("vc").sum().as("vc_sum"));
        
        
        // tEnv.toDataStream(table, WaterSensor.class).print();
    
       
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
